package com.ymh.spring.mq;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 
 * 名称：AsyncEchoServer<br/>
 * 描述：异步I/O模式<br/>
 * 创建时间：2018年3月12日 下午4:55:10 <br/>
 * 
 * @author aivin
 * @version 1.0.0
 */
public class AsyncEchoServer {

	public static int DEFAULT_PORT = 7;

	public static void main(String[] args) {
		int port;
		try {
			port = Integer.parseInt(args[0]);
		} catch (NumberFormatException e) {
			port = DEFAULT_PORT;
		}
		
		ExecutorService taskExecutor = Executors.newCachedThreadPool(Executors.defaultThreadFactory());
		//创建异步服务器socketChannel并绑定到默认组
		try(AsynchronousServerSocketChannel asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open()){
			if(asynchronousServerSocketChannel.isOpen()){
				//设置一些参数
				asynchronousServerSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4*1024);
				asynchronousServerSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
				
				//绑定服务器socket channel到本地地址
				asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
				
				//显示等待客户端的消息
				System.out.println("Waiting for connections ...");
				
				while(true) {
					Future<AsynchronousSocketChannel> asyncFuture = asynchronousServerSocketChannel.accept();
					
					try {
						final AsynchronousSocketChannel asynchronousSocketChannel = asyncFuture.get();
						Callable<String> worker = new Callable<String>() {

							@Override
							public String call() throws Exception {
								String host = asynchronousSocketChannel.getRemoteAddress().toString();
								
								System.out.println("Incoming connection from:"  + host);
								final ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
								//发送数据
								while(asynchronousSocketChannel.read(buffer).get()!=-1){
									buffer.flip();
									asynchronousSocketChannel.write(buffer).get();
									if(buffer.hasRemaining()){
										buffer.compact();
									}else {
										buffer.clear();
									}
								}
								asynchronousSocketChannel.close();
								System.out.println(host + "was successfully served");
								return host;
							}
						};
						taskExecutor.submit(worker);
					} catch (InterruptedException | ExecutionException e) {
						System.out.println(e);
						System.out.println("\n Server is shutting down...");
						//执行器不再接收新线程
						//并完成队列中所有的线程
						taskExecutor.shutdown();
						
						//等待所有线程完成
						while(! taskExecutor.isTerminated()) {
							
						}
						break;
					}
				}
			}else {
				System.out.println("The asynchronous server-socket channel cannot be opened!");
			}
		} catch (IOException e) {
			System.out.println(e);
		}
	}
}
